package com.ywu.mq.sdk.producer;

import com.ywu.boot.context.YwuContext;
import com.ywu.boot.context.YwuContextHolder;
import com.ywu.common.result.CommonResult;
import com.ywu.common.utils.TimerIdGenerateUtil;
import com.ywu.mq.sdk.config.YwuKafkaSdkConfig;
import com.ywu.mq.spi.entity.YwuKafkaProduceMessageEntity;
import com.ywu.mq.spi.spi.YwuKafkaSpi;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Properties;

/**
 * @ClassName YwuKafkaProducer
 * @Description kafka生产者模板方法
 * @Author GroundDemo
 * @Date 2025/3/28 9:51
 * @Version 1.0
 **/
@Component
@Slf4j
public abstract class YwuKafkaProducer {
    private YwuKafkaSdkConfig sdkConfig;

    @Autowired
    private YwuKafkaSpi kafkaSpi;

    protected static final Properties props = new Properties();

    public void init(YwuKafkaSdkConfig config) {
        this.sdkConfig = config;
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapService());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, config.getKeySerializer());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,config.getValueSerializer());
    }

    /**
    * @Author GroundDemo
    * @Description kafka生产的key
    * @Date 10:00 2025/3/28
    * @Param []
    * @return java.lang.String
    **/
    public abstract String getKey();

    /**
     * @Author GroundDemo
     * @Description kafka生产的value
     * @Date 10:00 2025/3/28
     * @Param []
     * @return java.lang.String
     **/
    public abstract String getValue();

    /**
     * @Author GroundDemo
     * @Description kafka生产的topic
     * @Date 10:00 2025/3/28
     * @Param []
     * @return java.lang.String
     **/
    public String getTopic() {
        return "";
    }

    /**
    * @Author GroundDemo
    * @Description 应用ID
    * @Date 12:29 2025/3/29
    * @Param []
    * @return java.lang.String
    **/
    public String getAppId() {
        return "";
    }

    /**
    * @Author GroundDemo
    * @Description 自定义生产回调
    * @Date 10:51 2025/3/28
    * @Param [metadata, exception, messageId]
    * @return void
    **/
    public abstract void customCallBack(RecordMetadata metadata, Exception exception, String messageId);

    /**
    * @Author GroundDemo
    * @Description 发送后的回调
    * @Date 10:50 2025/3/28
    * @Param [metadata, exception, messageId]
    * @return void
    **/
    private void doProduceCallBack(RecordMetadata metadata, Exception exception, String messageId, YwuContext currentContext) {
        YwuContextHolder.setCurrentContext(currentContext);
        // 调用spi记录生产日志
        YwuKafkaProduceMessageEntity entity = new YwuKafkaProduceMessageEntity();
        entity.setId(messageId);
        if (exception != null) {
            log.error("produce error {}", exception.getMessage());
            entity.setStatus(YwuKafkaProduceMessageEntity.PRODUCE_ERROR);
        } else {
            entity.setStatus(YwuKafkaProduceMessageEntity.PRODUCE_SUCCESS);
            log.debug("produce send success partition {}, offset {}", metadata.partition(), metadata.offset());
        }
        kafkaSpi.produceUpdateMessage(entity);
        // 执行自定义回调
        customCallBack(metadata, exception, messageId);
    }

    /**
    * @Author GroundDemo
    * @Description 生产消息
    * @Date 9:57 2025/3/28
    * @Param []
    * @return void
    **/
    public void produce() {
        YwuContext currentContext = YwuContextHolder.getCurrentContext();
        // 创建生产者实例
        try (Producer<String, String> producer = new KafkaProducer<>(props)) {
            // 生产的值和key
            String key = getKey();
            String value = getValue();
            // 要生产到的主题
            String topic = getTopic();
            if (StringUtils.isEmpty(topic)) {
                topic = sdkConfig.getTopicName();
            }
            String topicName = "";
            if (StringUtils.isEmpty(getAppId())) {
                String appId = sdkConfig.getTopicAppId();
                String replaceAppId = appId.replace(".", "_").replace("-", "_");
                topicName = replaceAppId + "-" + topic;
            } else {
                String appId = getAppId();
                String replaceAppId = appId.replace(".", "_").replace("-", "_");
                topicName = replaceAppId + "-" + topic;
            }
            // 构建对象
            YwuKafkaProduceMessageEntity ywuKafkaProduceMessageEntity = new YwuKafkaProduceMessageEntity();
            ywuKafkaProduceMessageEntity.setTopicName(topic);
            ywuKafkaProduceMessageEntity.setId(TimerIdGenerateUtil.nextId());
            ywuKafkaProduceMessageEntity.setKey(key);
            ywuKafkaProduceMessageEntity.setValue(value);
            ywuKafkaProduceMessageEntity.setStatus(YwuKafkaProduceMessageEntity.PRODUCING);
            ywuKafkaProduceMessageEntity.setAppId(this.sdkConfig.getAppId());
            ywuKafkaProduceMessageEntity.setTopicAppId(this.sdkConfig.getTopicAppId());
            ywuKafkaProduceMessageEntity.setAppKey(this.sdkConfig.getAppKey());
            // 调用spi查询主题是否存在并进行权限验证
            CommonResult<Object> checkRes = kafkaSpi.produceCheckTopic(ywuKafkaProduceMessageEntity);
            if (!checkRes.getCode().equals("200")) {
                log.error("topic is not exit OR topic permission is refused");
                throw new RuntimeException("topic is not exit OR topic permission is refused");
            }
            // 调用spi上报日志
            CommonResult<Object> result = kafkaSpi.produceCreateMessage(ywuKafkaProduceMessageEntity);
            if (!result.getCode().equals("200")) {
                log.error("mq spi error {}", result.getMessage());
                throw new RuntimeException("mq spi error");
            }
            // 构建新的value值，将消息ID存入value中 - 后续消费的时候会将其进行切割
            value = ywuKafkaProduceMessageEntity.getId() + "-ywuMessageId-" + value;
            // 构建ProducerRecord对象
            ProducerRecord<String, String> record = new ProducerRecord<>(topicName, key, value);
            producer.send(record, (metadata, exception) -> {
                // 生产回调
                doProduceCallBack(metadata, exception, ywuKafkaProduceMessageEntity.getId(), currentContext);
            });
        }
    }
}
